package org.atguigu.gmall.realtime.util;

import org.apache.commons.lang.math.RandomUtils;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Random;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.util
 * Author: ZengHaiFeng
 * CreateTime:2023/4/19 16:58:11
 * Description: Todo
 * Version：1.0
 */
public class FlinkSourceUtil {
    public static KafkaSource<String> getKafkaSource(String bootStrapServer,String topic,String groupId){
        return KafkaSource.<String>builder()
                .setBootstrapServers(bootStrapServer)
                .setStartingOffsets(OffsetsInitializer.latest())
                .setTopics(topic)
                .setGroupId(groupId)
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null){
                            return new String(message, StandardCharsets.UTF_8);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(new TypeHint<String>(){});
                    }
                })
                .setClientIdPrefix(groupId + "-")
                .setProperty("isolation.level","read_committed")
                .build();
    }
}
